package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.CourseOrderBean;
import com.atguigu.utils.DateFormatUtil;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;

public class Dws131_CourseOrderWindow {
    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取KafkaDWD层下单明细主题数据
        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("dwd_trade_order_detail", "course_order"), WatermarkStrategy.noWatermarks(), "kafka-source");

        //3.过滤Null值并转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    out.collect(JSON.parseObject(value));
                }
            }
        });

        jsonObjDS.print("jsonObjDS--->");

        //4.按照订单明细ID分组,去重由Left Join产生的重复数据，并直接转换为JavaBean对象
        SingleOutputStreamOperator<CourseOrderBean> courseOrderDS = jsonObjDS
                .keyBy(json -> json.getString("id"))
                .flatMap(new RichFlatMapFunction<JSONObject, CourseOrderBean>() {

                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        StateTtlConfig ttlConfig = new StateTtlConfig.Builder(Time.seconds(5))
                                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                                .build();
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("value-state", String.class);
                        stateDescriptor.enableTimeToLive(ttlConfig);

                        valueState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public void flatMap(JSONObject value, Collector<CourseOrderBean> out) throws Exception {

                        //取出状态数据
                        String state = valueState.value();

                        if (state == null) {
                            valueState.update("1");

                            HashSet<String> orderIds = new HashSet<>();
                            orderIds.add(value.getString("order_id"));

                            out.collect(new CourseOrderBean(
                                    "",
                                    "",
                                    value.getString("course_id"),
                                    value.getString("course_name"),
                                    orderIds,
                                    value.getString("user_id"),
                                    0L,
                                    value.getBigDecimal("origin_amount"),
                                    value.getLong("create_time")));
                        }
                    }
                });
        courseOrderDS.print("courseOrderDS-->");

        //5.按照UID分组
        KeyedStream<CourseOrderBean, String> keyedByUidDS = courseOrderDS.keyBy(CourseOrderBean::getUserID);

        //6.去重UID
        SingleOutputStreamOperator<CourseOrderBean> courseOrderByUidDS = keyedByUidDS.map(new RichMapFunction<CourseOrderBean, CourseOrderBean>() {

            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("uid-state", String.class);

                StateTtlConfig ttlConfig = new StateTtlConfig
                        .Builder(Time.seconds(10))
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                        .build();
                stateDescriptor.enableTimeToLive(ttlConfig);

                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public CourseOrderBean map(CourseOrderBean value) throws Exception {

                //获取状态数据
                String state = valueState.value();

                if (state == null) {
                    value.setCourseOrderUserCount(1L);
                    valueState.update("1");
                }

                return value;
            }
        });

        courseOrderByUidDS.print("courseOrderByUidDS--->");

        //7.提取时间戳生成Watermark
        SingleOutputStreamOperator<CourseOrderBean> courseOrderWithWMDS = courseOrderByUidDS.assignTimestampsAndWatermarks(WatermarkStrategy.<CourseOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<CourseOrderBean>() {
            @Override
            public long extractTimestamp(CourseOrderBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        //8.分组、开窗、聚合
        SingleOutputStreamOperator<CourseOrderBean> reduceDS = courseOrderWithWMDS.keyBy(CourseOrderBean::getCourseId)
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<CourseOrderBean>() {
                    @Override
                    public CourseOrderBean reduce(CourseOrderBean value1, CourseOrderBean value2) throws Exception {
                        value1.setCourseOrderAmount(value1.getCourseOrderAmount().add(value2.getCourseOrderAmount()));
                        value1.setCourseOrderUserCount(value1.getCourseOrderUserCount() + value2.getCourseOrderUserCount());

                        return value1;
                    }
                }, new WindowFunction<CourseOrderBean, CourseOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<CourseOrderBean> input, Collector<CourseOrderBean> out) throws Exception {

                        CourseOrderBean next = input.iterator().next();

                        next.setTs(System.currentTimeMillis());
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));

                        out.collect(next);
                    }
                });

        reduceDS.print("reduceDS-->");

        //9.将数据写出
        reduceDS.print("reduceDS>>>>>>>>>");
        reduceDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_course_order_window values(?,?,?,?,?,?,?)"));

        //10.启动任务
        env.execute("Dws131_CourseOrderWindow");


    }
}
